/*
 * $AIST_Release: 0.9.0 $
 * Copyright 2011 Information Technology Research Institute, National
 * Institute of Advanced Industrial Science and Technology
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package aggr

import Aggregators._
import scala.collection.mutable.{Map => MMap}
import scala.collection.JavaConversions._

import com.google.protobuf.Descriptors._
import com.google.protobuf.Descriptors.FieldDescriptor.Type
import com.google.protobuf.Message
import com.google.protobuf.DynamicMessage

import util.JavaTypes._
import SerializedTypes._
import scala.collection.mutable.PriorityQueue
import Util._

class PriorityContainer(capacity: Long, cmp: (Any, Any) => Int) {
  var size = 0
  var data: PriorityQueue[(Any, Any)] = new PriorityQueue()(new Ordering[(Any, Any)] {
    def compare(x: (Any, Any), y: (Any, Any)): Int = cmp(y._2, x._2) /* Reverse */
  })

  def add(value: (Any, Any)): Unit = {
    data.enqueue(value)
    if (data.length > capacity) {
      data.dequeue()
    }
  }
  def result() = data.toList.map { _._1 }
}

/**
 * "maximum" aggregator.
 */
class Maximum(valType: SerializedType, N: Long) extends Aggregator {
    type Tuple = MMap[FieldDescriptor, Any]
    def emptyTuple: Tuple = MMap()

    val MessageType(valDesc) = valType
    val valFld    = valDesc.findFieldByName("data__")
    val weightFld = valDesc.findFieldByName("weight")

    def compare(lhs: Any, rhs: Any): Int = {
      (lhs, rhs) match {
        case (l: JDouble, r: JDouble) => l.doubleValue.compare(r.doubleValue)
        case (l: JFloat,  r: JFloat)  => l.floatValue.compare(r.floatValue)
        case (l: JInt,    r: JInt)    => l.intValue.compare(r.intValue)
        case (l: JLong,   r: JLong)   => l.longValue.compare(r.longValue)
        case (l: Message, r: Message) => 
          def f(l: List[Any], r: List[Any]): Int = (l, r) match {
            case (l::ls, r::rs) => 
              compare(l, r) match {
                case 0 => f(ls, rs)
                case x => x
              }
            case (Nil, Nil) => 0
            case _ => throw new IllegalArgumentException()
          }
          f(l.getAllFields.values.toList, r.getAllFields.values.toList)
      }
    }

    override def aggregate(data: TraversableOnce[(Any, Any)])(writer: (Any, Any) => Unit): Unit = {

      var m: MMap[Any, PriorityContainer] = MMap()

      data.foreach { case (k, v) =>
        m.getOrElseUpdate(k, new PriorityContainer(N, compare)).add( v match {
          case m:Message => (m.getField(weightFld), m.getField(valFld))
        })
      }

      m.foreach { case (k, v) =>
        v.result.foreach { x => 
          writer(k, x)
        }
      }
    }

    override def getValueType(): Option[SerializedType] = Some(SerializedTypes.getType(valFld))
}


class MaximumFactory extends AggregatorFactory {

    override def createNew(
                  keyType: SerializedType,
                  valType: SerializedType,
                  args: List[Long]): Aggregator = {
      new Maximum(valType, args(0))
    }

    override def check(keyType: SerializedType,
              valType: SerializedType,
              args: List[SerializedType], 
              options: Map[String, SerializedType]): Option[List[String]]= {

    val aggrName = "maximum"

    val argsErrors = checkNumOfArgs(aggrName, args.length, 1) match {
      case Nil if (args(0) != Int64Type) => List("First arguments must be int type")
      case x => x
    }

    val optionErrors = checkOptionName(aggrName, options, List("weight")) match {
      case Nil if (!isNumber(options("weight"))) => List("\"weight\" option must be countable.")
      case x   => x
    }

    (argsErrors ++ optionErrors) match {
      case Nil => None
      case x   => Some(x)
    }
  }
}
